package State

import Source.SensorReading
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.time.Time
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

import java.util.concurrent.TimeUnit


object CheckPointTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    /**
     * 检查点的配置
     */
    //默认是不启用checkpoint的，这里首先需要先打开checkpoint
    env.enableCheckpointing(1000l) //时间间隔是1000毫秒,默认是500l
    env.getCheckpointConfig.setCheckpointInterval(500l) //重新设置间隔为500毫秒
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE) //至少一次，默认是EXACTLY_ONCE（精确一次）
    env.getCheckpointConfig.setCheckpointTimeout(60 * 1000l) //超时时间
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(4) //最大并行的checkPoint,默认是1
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(100l) //最小的间隔时间
    env.getCheckpointConfig.setPreferCheckpointForRecovery(true) //是否更加倾向于使用checkPoint来做故障恢复
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2) //最多认容多少次的checkpoint失败

    /**
     * 配置重启策略
     * failureRateRestart 失败率重启
     * failBackRestart 失败回滚重启
     * fixedDelayRestart 固定时间重启
     * noRestart 不重启(默认)
     */
    //重启三次，每次间隔1秒
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000l))

    //在5分钟内失败5次，每次重启的时间间隔是10秒
    env.setRestartStrategy(RestartStrategies.failureRateRestart(5,
      Time.of(5, TimeUnit.MINUTES), Time.of(10, TimeUnit.SECONDS)))

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }).keyBy(_.id)
  }
}
